package demo;

import java.util.List;
import java.util.concurrent.TimeUnit;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageQueue;

/**
 * 生产者
 * 
 * @author super
 *
 */
public class Producer2 {

	public static void main(String[] args) throws Exception{

		/*
		 * 一个应用创建一个Producer，由应用来维护此对象，可以设置为全局对象或者单例
		 * 注意：ProducerGroupName需要由应用来保证唯一
		 * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
		 * 因为服务器会回查这个Group下的任意一个Producer
		 */
		DefaultMQProducer producer = new DefaultMQProducer(Common.PRODUCER_GROUP_NAME);
		producer.setNamesrvAddr(Common.ADDR);
		
		/*
		 * 默认情况下不需要设置instanceName，rocketmq会使用ip@pid作为instanceName(pid代表jvm名字)
		 * 如果设置instanceName，rocketmq会使用ip@instanceName作为consumer的唯一标示，此时需要注意instanceName需要不同
		 */
		//producer.setInstanceName("Producer");

		
		// Producer对象在使用之前必须要调用start初始化，初始化一次即可。注意：切记不可以在每次发送消息时，都调用start方法
		producer.start();

		// producer.setCreateTopicKey("OrderID001");
		// producer.createTopic("OrderID001", "TopicTest1", 0);

		/*
		 * 下面这段代码表明一个Producer对象可以发送多个topic，多个tag的消息。
		 * 注意：send方法是同步调用，只要不抛异常就标识成功。但是发送成功也可会有多种状态，
		 * 例如消息写入Master成功，但是Slave不成功，这种情况消息属于成功，但是对于个别应用如果对消息可靠性要求极高，
		 * 需要对这种情况做处理。另外，消息可能会存在发送失败的情况，失败重试由应用来处理。
		 */
		for (int i = 0; i < 6; i++) {

			// 普通消息

			// Message参数topic、tag、keys、body
			String body = "Hello MetaQ TopicTest1 TagA OrderID00" + i;
			Message msg = new Message("TopicTest1", "TagA", "OrderID00" + i, (body).getBytes());

			SendResult sendResult = sendMsg(producer, msg);
			System.out.println(JSON.toJSONString(sendResult));
			
			

			// 顺序消息
			/*String body2 = "Hello MetaQ TopicTest2 TagA OrderID00" + i;
			Message msg2 = new Message("TopicTest2", "TagA", "OrderID00" + i, (body2).getBytes());

			SendResult sendResult2 = sendMsg2(producer, msg2);
			System.out.println(JSON.toJSONString(sendResult2));*/

			TimeUnit.MILLISECONDS.sleep(50);
		}

		/**
		 * 应用退出时，要调用shutdown来清理资源，关闭网络连接，从MetaQ服务器上注销自己
		 * 注意：我们建议应用在JBOSS、Tomcat等容器的退出钩子里调用shutdown方法
		 */
		producer.shutdown();
	}
	
	
	/**
	 * 并行消息(默认消息)
	 * 不追求顺序，只要结果消息消费掉，效率高
	 * 
	 * @param producer
	 * @param msg
	 * @throws Exception
	 */
	static SendResult sendMsg(DefaultMQProducer producer, Message msg) throws Exception {
		System.out.println("发送消息：" + JSON.toJSONString(msg));
		return producer.send(msg);

	}
	
	/**
	 * 顺序消息(串行)
	 * 
	 * 严格按照时间消费的模式，这种模式需要用串行方式，生产者生产的时候，这时候生产者需要往特定的队列里有序push。
	 * 消费者也要按照顺序严格有序消费，用这个有序的监听者ConsumeOrderlyContext
	 * 
	 * @param producer
	 * @param msg
	 * @throws Exception
	 */
	static SendResult sendMsg2(DefaultMQProducer producer, Message msg) throws Exception {

		// 指定规则
		return producer.send(msg, new MessageQueueSelector() {
			@Override
			public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
				Integer id = arg.hashCode();
				int index = id % mqs.size();
				return mqs.get(index);
			}
		}, "Type");
		
	}
	
	
}
